package com.study.ws.config.mq;

import com.study.ws.util.executor.TheadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.util.ObjectUtils;

/**
 * @author xl-9527
 * @since 2025/3/29
 **/
@Configuration(proxyBeanMethods = false)
public class RabbitMQConfiguration {

    public final static String TOPIC_QUEUENAME = "ws-mq-topic";

    private static final Logger log = LoggerFactory.getLogger(RabbitMQConfiguration.class);

    private final SimpMessagingTemplate simpMessagingTemplate;

    public RabbitMQConfiguration(final SimpMessagingTemplate simpMessagingTemplate) {
        this.simpMessagingTemplate = simpMessagingTemplate;
    }

    @Bean
    Queue topicQueueWithWsStomp() {
        return QueueBuilder.durable(TOPIC_QUEUENAME).build();
    }

    @Bean
    Exchange topicExchange() {
        return ExchangeBuilder.topicExchange(TOPIC_QUEUENAME).durable(true).build();
    }

    @Bean
    Binding binding(final Queue topicQueueWithWsStomp, final Exchange topicExchange) {
        return BindingBuilder.bind(topicQueueWithWsStomp).to(topicExchange).with(TOPIC_QUEUENAME).noargs();
    }

    @Bean
    ConnectionFactory connectionFactory(final CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer) {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        cachingConnectionFactoryConfigurer.configure(factory);

        // custom conf
        // - virtual host
        factory.setVirtualHost("/ws/topic");
        return factory;
    }

    @Bean
    RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }

    @Bean
    SimpleMessageListenerContainer simpleMessageListenerContainer(final ConnectionFactory connectionFactory) {
        final SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        // listener topic
        listenerContainer.setQueueNames(TOPIC_QUEUENAME);
        listenerContainer.setConcurrency("2");
        // set ack mode
        listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        listenerContainer.setTaskExecutor(TheadPoolUtil.getTaskExecutor(2, "ws-mq-topic"));
        // set listener
        listenerContainer.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            try {
                final String msg = new String(message.getBody());
                simpMessagingTemplate.convertAndSend("/ws/mq/topic", msg);
                log.info("receive msg：{} from rabbitmq", msg);
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                if (!ObjectUtils.isEmpty(channel)) {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
                }
            }
        });
        return listenerContainer;
    }
}
